package design_pattern.publish_subscribe;

import threadpool.ThreadPoolFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

public class Broker<M> {

    //订阅器名称
    private String name;

    //订阅器队列容量
    private static final int QUEUE_CAPACITY = 5;

    //订阅存储队列
    private Queue<Message<M>> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    //订阅者
    private List<Subscriber<M>> subscribers = new ArrayList<>();

    public Broker(String name) {
        this.name = name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    /**
     * @param publisher 发布者
     * @param msg       消息，为泛型，可能为 String List Map 等
     * @param isInstant 是否为即时消息。
     *                  如果为即时消息，则马上通知订阅者。
     *                  如果不是即时消息，则将消息放入队列。在队列未满的情况下，再以FIFO的顺序通知订阅者。
     */
    public void publish(String publisher, M msg, boolean isInstant) {
        if (isInstant) {
            listen(publisher, msg);
            return;
        }

        Message<M> message = new Message<>(publisher, msg);
        if (!queue.offer(message)) {
            System.out.println("队列已满");
            listen();
            return;
        }

        ThreadPoolFactory.getThreadPool().execute(new Runnable() {
            @Override
            public void run() {
                release();
            }
        });
    }

    public void subscribe(Subscriber<M> subscriber) {
        subscribers.add(subscriber);
    }

    public void unSubscribe(Subscriber<M> subscriber) {
        subscribers.remove(subscriber);
    }

    private void listen() {
        Message<M> message;
        if ((message = queue.poll()) != null) {
            listen(message.getPublisher(), message.getMsg());
        }
    }

    private void listen(String publisher, M msg) {
        for (Subscriber<M> subscriber : subscribers) {
            subscriber.listen(publisher, msg);
        }
    }

    private void release() {
        Message<M> message;
        while ((message = queue.poll()) != null) {
            listen(message.getPublisher(), message.getMsg());
        }
    }

}
